package temp.casia.isiteam.zrb.hbase.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import temp.casia.isiteam.zrb.hbase.client.bean.HBCityComparison;
import temp.casia.isiteam.zrb.hbase.client.bean.HBCityEconomic;
import temp.casia.isiteam.zrb.hbase.client.bean.HBCityIndustry;
import temp.casia.isiteam.zrb.hbase.client.bean.HBCityInfo;
import temp.casia.isiteam.zrb.hbase.client.bean.HbaseStatus;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableCityComparison;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableCityEconomic;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableCityIndustry;
import temp.casia.isiteam.zrb.hbase.client.bean.table.HbaseTableCityInfo;
import temp.casia.isiteam.zrb.hbase.client.util.Commons;
import temp.casia.isiteam.zrb.hbase.client.util.ConnectionManager;
import temp.casia.isiteam.zrb.hbase.client.util.MD5Util;

/**
 * 城市信息操作类<br>
 * 提供HBase中城市相关表的查询、写入
 * 
 * @author 彭鑫
 * @createDate 2015年6月29日
 */
public class CityHbaseClient {
	protected Configuration config;
	private Logger logger = Logger.getLogger(CityHbaseClient.class);
	protected String quorum;
	/**
	 * 表名
	 */
	protected Map<String, TableName> tableNameMap;
	protected Connection conn;
	protected Integer currentClientNum = 0;
	private Timer timer;

	/**
	 * 使用默认的master地址
	 */
	public CityHbaseClient(){
		this.quorum = Commons.quorum;
		this.initConfig();
		this.startTimer();
	}
	
	/**
	 * 指定master地址
	 * @param quorum
	 */
	public CityHbaseClient(String quorum){
		this.quorum = quorum;
		this.initConfig();
		this.startTimer();
	}
	
	protected void initConn() {
		this.conn = ConnectionManager.getConn(config);
	}
	
	protected void initConfig() {
		try {
			config = HBaseConfiguration.create();
			config.set("hbase.zookeeper.quorum", this.quorum);
			config.set("hbase.client.retries.number", "3");
			config.set("zookeeper.recovery.retry", "3");
			
			conn = ConnectionManager.getConn(config);
			Admin admin = conn.getAdmin();
			TableName[] tableAry = admin.listTableNames("city_.*");
			this.tableNameMap = new HashMap<String, TableName>();
			for (TableName tableName : tableAry) {
				this.tableNameMap.put(tableName.getNameAsString(), tableName);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	/**
	 * 定时任务，超过一定时间不使用连接，则自动关闭
	 */
	private void startTimer(){
		long delay = 2 * 1000;
		long period = 10 * 1000;
		this.timer = new Timer();
		this.timer.schedule(new TimerTask() {
			@Override
			public void run() {
				timerTask();				
			}
		}, delay, period);
	}

	private void timerTask(){
		synchronized (this.currentClientNum) {
			if(this.currentClientNum == 0){
				this.closeConn();
			}
		}
	}
	
	/**
	 * 查询城市所有信息
	 * 
	 * @param postcode
	 * @return
	 * @throws Exception
	 */
	public HBCityInfo getCityInfoAll(String postcode) {
		HBCityInfo cityInfo = this.getCityInfoBasic(postcode);
		if (cityInfo != null) {
			List<HBCityEconomic> economicList = this
					.getCityInfoAllEconomic(postcode);
			cityInfo.setEconomicList(economicList);
			List<HBCityIndustry> industryList = this.getCityIndustry(postcode);
			cityInfo.setIndustryList(industryList);
		}

		return cityInfo;
	}

	/**
	 * 查询城市基本信息
	 * 
	 * @param postcode
	 * @return
	 */
	public HBCityInfo getCityInfoBasic(String postcode) {
		this.currentClientNum++;
		HBCityInfo cityInfo = null;
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap
					.get(HbaseTableCityInfo.TABLE_NAME));
			Get get = new Get(postcode.getBytes());
			Result result = table.get(get);
			if (result != null) {
				cityInfo = this.getCityEntity(result);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;

		return cityInfo;
	}
	
	/**
	 * 批量查询城市
	 * @param postcodeList
	 * @return
	 */
	public List<HBCityInfo> getCityInfoBasic(List<String> postcodeList){
		this.currentClientNum++;
		List<HBCityInfo> cityList = new ArrayList<HBCityInfo>();
		Table table = null;
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap
					.get(HbaseTableCityInfo.TABLE_NAME));
			List<Get> getList = new ArrayList<Get>();
			for(String postcode: postcodeList){
				Get get = new Get(Bytes.toBytes(postcode));
				getList.add(get);
			}
			Result[] results = table.get(getList);
			for(Result result: results){
				HBCityInfo city = this.getCityEntity(result);
				cityList.add(city);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return cityList;
	}
	
	private HBCityInfo getCityEntity(Result result){
		HBCityInfo cityInfo = new HBCityInfo();
		cityInfo.setCnName(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_cnName));
		cityInfo.setEnName(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_enName));
		cityInfo.setOtherName(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_otherName));
		cityInfo.setRegionType(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_regionType));
		cityInfo.setProvince(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_province));
		cityInfo.setTopCity(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_topCity));
		cityInfo.setZfzd(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_zfzd));
		cityInfo.setLongitude(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_longitude));
		cityInfo.setLatitude(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_latitude));
		cityInfo.setAreacode(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_areacode));
		cityInfo.setPostcode(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_postcode));
		cityInfo.setAre(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_are));
		cityInfo.setRegionCode(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_regionCode));
		cityInfo.setGeoPosition(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_geoPosition));
		cityInfo.setNearby(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_nearby));
		cityInfo.setPopulation(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_population));
		cityInfo.setRegionFeature(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_regionFeature));
		cityInfo.setLandTran(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_landTran));
		cityInfo.setShipping(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_shipping));
		cityInfo.setWaterway(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_waterway));
		cityInfo.setTraffic(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_traffic));
		cityInfo.setSite(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_site));
		cityInfo.setUrl(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_url));
		cityInfo.setClimate(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_climate));
		cityInfo.setPlateNum(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_plateNum));
		cityInfo.setImg(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_img));
		cityInfo.setClimate(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_climate));
		cityInfo.setPlateNum(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_plateNum));
		cityInfo.setImg(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_img));
		cityInfo.setVideoUrl(this.getRow(result, HbaseTableCityInfo.COLUMN_FAMILY, HbaseTableCityInfo.Column_Name_videoUrl));
		
		return cityInfo;
	}

	/**
	 * 查询城市全部经济信息
	 * 
	 * @param postcode
	 * @return
	 */
	public List<HBCityEconomic> getCityInfoAllEconomic(String postcode){
		this.currentClientNum++;
		List<HBCityEconomic> ecoList = new ArrayList<HBCityEconomic>();
		Table table = null;
		ResultScanner rs = null;
		
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityEconomic.TABLE_NAME));
			String startRow = postcode + "_eco_0";
			String stopRow = postcode + "_eco_2100";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBCityEconomic ce = this.getCityEconomicEntity(result);
				String rowKey = Bytes.toString(result.getRow());
				ce.setYear(Integer.parseInt(rowKey.replace(postcode + "_eco_", "")));
				ce.setPostCode(postcode);
				ecoList.add(ce);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(table, conn);		}
		this.currentClientNum--;
		return ecoList;
	}

	/**
	 * 查询指定年份城市经济数据
	 * 
	 * @param postcode
	 * @param comCityPostcode
	 * @return
	 */
	public HBCityEconomic getCityInfoEconomic(String postcode, int year) {
		this.currentClientNum++;
		Table table = null;
		HBCityEconomic ce = null;
		
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityEconomic.TABLE_NAME));
			String rowKey = postcode + "_eco_" + year;
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			if(result != null){
				ce = this.getCityEconomicEntity(result);
				ce.setPostCode(postcode);
				ce.setYear(year);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ce;
	}
	
	/**
	 * 将查询结果封装
	 * @param result
	 * @return
	 */
	protected HBCityEconomic getCityEconomicEntity(Result result){
		HBCityEconomic ce = new HBCityEconomic();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableCityEconomic.Column_Name_gdp)){ce.setGdp(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_fixedInvest)){ce.setFixedInvest(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_pcdi)){ce.setPcdi(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_gdpGRate)){ce.setGdpGRate(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_publicBR)){ce.setPublicBR(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_publicBRGRate)){ce.setPublicBRGRate(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_fai)){ce.setFai(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_faiGRate)){ce.setFaiGRate(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_rscg)){ce.setRscg(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_rscgGRate)){ce.setRscgGRate(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_jmrjkzpsr)){ce.setJmrjkzpsr(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_jmrjkzpsrzzl)){ce.setJmrjkzpsrzzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_czjmrjkzpsr)){ce.setCzjmrjkzpsr(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_czjmrjkzpsrzzl)){ce.setCzjmrjkzpsrzzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_ncjmrjkzpsr)){ce.setNcjmrjkzpsr(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_ncjmrjkzpsrzzl)){ce.setNcjmrjkzpsrzzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_czjmzl)){ce.setCzjmzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_ncjmzl)){ce.setNcjmzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy1zzl)){ce.setCy1zzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy1zjz)){ce.setCy1zjz(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy1zzcy)){ce.setCy1zzcy(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy1js)){ce.setCy1js(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy2zzl)){ce.setCy2zzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy2zjz)){ce.setCy2zjz(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy2zzcy)){ce.setCy2zzcy(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy2js)){ce.setCy2js(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy3zzl)){ce.setCy3zzl(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy3zjz)){ce.setCy3zjz(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy3zzcy)){ce.setCy3zzcy(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_cy3js)){ce.setCy3js(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_dfczzsr)){ce.setDfczzsr(value);}
			else if(quilifier.equals(HbaseTableCityEconomic.Column_Name_dfczzsrzzl)){ce.setDfczzsrzzl(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return ce;
	}

	/**
	 * 获取城市所有产业信息
	 * 
	 * @param postcode
	 * @return
	 */
	public List<HBCityIndustry> getCityIndustry(String postcode) {
		this.currentClientNum++;
		List<HBCityIndustry> ecoList = new ArrayList<HBCityIndustry>();
		Table table = null;
		ResultScanner rs = null;
		
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityIndustry.TABLE_NAME));
			String startRow = postcode + "_ind_0";
			String stopRow = postcode + "_ind_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBCityIndustry ce = this.getCityIndustryEntity(result);
				ce.setPostCode(postcode);
				ecoList.add(ce);
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ecoList;
	}

	protected HBCityIndustry getCityIndustryEntity(Result result){
		HBCityIndustry ci = new HBCityIndustry();
		for (Cell cell : result.rawCells()) {
			String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
			String value = Bytes.toString(CellUtil.cloneValue(cell));
			//只有一个family，因此不做family的判断
			if(quilifier.equals(HbaseTableCityIndustry.Column_Name_name)){ci.setName(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_attractInvest)){ci.setAttractInvest(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_plan)){ci.setPlan(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_police)){ci.setPolice(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_leader)){ci.setLeader(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_cybj)){ci.setCybj(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_advantage)){ci.setAdvantage(value);}
			else if(quilifier.equals(HbaseTableCityIndustry.Column_Name_cyzc)){ci.setCyzc(value);}
			else {
				this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
			}
		}
		return ci;
	}
	
	/**
	 * 查询对比城市信息
	 * 
	 * @param postcode
	 * @return
	 */
	public List<HBCityComparison> getCityComInfo(String postcode) {
		this.currentClientNum++;
		List<HBCityComparison> ccList = new ArrayList<HBCityComparison>();
		Table table = null;
		ResultScanner rs = null;
		
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityComparison.TABLE_NAME));
			String startRow = postcode + "_com_0";
			String stopRow = postcode + "_com_z";
			Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
			rs = table.getScanner(scan);
			for(Result result: rs.next(100)){
				HBCityComparison ce = new HBCityComparison();
				for (Cell cell : result.rawCells()) {
					String quilifier = Bytes.toString(CellUtil.cloneQualifier(cell));
					String value = Bytes.toString(CellUtil.cloneValue(cell));
					if(quilifier.equals(HbaseTableCityComparison.Column_Name_name)){ce.setName(value);}
					else if(quilifier.equals(HbaseTableCityComparison.Column_Name_report)){ce.setReport(value);}
					else {
						this.logger.warn("unknown quilifier! quilifier: " + quilifier);;
					}
				}
				String rowKey = Bytes.toString(result.getRow());
				ce.setComPostCode(rowKey.replace(postcode + "_com_", ""));
				ce.setPostCode(postcode);
				ccList.add(ce);
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return ccList;
	}

	/**
	 * 提交城市信息
	 * 
	 * @param city
	 * @return status
	 */
	public HbaseStatus insertCityInfo(HBCityInfo city) {
		this.currentClientNum++;
		HbaseStatus status = HbaseStatus.successed;
		Table table = null;
		if(city.getPostcode() == null || city.getPostcode().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityInfo.TABLE_NAME));
			Put put = new Put(Bytes.toBytes(city.getPostcode()));
			byte[] famByt = Bytes.toBytes(HbaseTableCityInfo.COLUMN_FAMILY);
			
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_cnName, city.getCnName());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_enName, city.getEnName());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_otherName, city.getOtherName());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_regionType, city.getRegionType());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_province, city.getProvince());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_topCity, city.getTopCity());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_zfzd, city.getZfzd());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_longitude, city.getLongitude());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_latitude, city.getLatitude());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_areacode, city.getAreacode());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_postcode, city.getPostcode());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_are, city.getAre());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_regionCode, city.getRegionCode());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_geoPosition, city.getGeoPosition());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_nearby, city.getNearby());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_population, city.getPopulation());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_regionFeature, city.getRegionFeature());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_landTran, city.getLandTran());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_shipping, city.getShipping());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_waterway, city.getWaterway());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_traffic, city.getTraffic());	
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_site, city.getSite());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_url, city.getUrl());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_climate, city.getClimate());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_plateNum, city.getPlateNum());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_img, city.getImg());
			this.addValue(put, famByt, HbaseTableCityInfo.Column_Name_videoUrl, city.getVideoUrl());
			
			table.put(put);
			if(city.getEconomicList() != null && city.getEconomicList().size() > 0) {
				for(HBCityEconomic ce: city.getEconomicList()){
					this.insertCityEconomic(ce);
				}
			}
			if(city.getIndustryList() != null && city.getIndustryList().size() > 0) {
				for(HBCityIndustry ci: city.getIndustryList()){
					this.insertCitydustry(ci);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}

	/**
	 * 提交城市年度经济数据
	 * 
	 * @param city
	 * @return status
	 */
	public HbaseStatus insertCityEconomic(HBCityEconomic economic) {
		this.currentClientNum++;
		HbaseStatus status = HbaseStatus.successed;
		Table table = null;
		if(economic.getPostCode() == null || economic.getPostCode().length() == 0 || economic.getYear() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityEconomic.TABLE_NAME));
			Put put = new Put(Bytes.toBytes(economic.getPostCode() + "_eco_" + economic.getYear()));
			byte[] famByt = Bytes.toBytes(HbaseTableCityEconomic.COLUMN_FAMILY);
			
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_gdp, economic.getGdp());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_fixedInvest, economic.getFixedInvest());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_pcdi, economic.getPcdi());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_gdpGRate, economic.getGdpGRate());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_publicBR, economic.getPublicBR());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_publicBRGRate, economic.getPublicBRGRate());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_fai, economic.getFai());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_faiGRate, economic.getFaiGRate());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_rscg, economic.getRscg());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_rscgGRate, economic.getRscgGRate());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_jmrjkzpsr, economic.getJmrjkzpsr());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_jmrjkzpsrzzl, economic.getJmrjkzpsrzzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_czjmrjkzpsr, economic.getCzjmrjkzpsr());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_czjmrjkzpsrzzl, economic.getCzjmrjkzpsrzzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_ncjmrjkzpsr, economic.getNcjmrjkzpsr());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_ncjmrjkzpsrzzl, economic.getNcjmrjkzpsrzzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_czjmzl, economic.getCzjmzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_ncjmzl, economic.getNcjmzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy1zzl, economic.getCy1zzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy1zjz, economic.getCy1zjz());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy1zzcy, economic.getCy1zzcy());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy1js, economic.getCy1js());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy2zzl, economic.getCy2zzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy2zjz, economic.getCy2zjz());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy2zzcy, economic.getCy2zzcy());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy2js, economic.getCy2js());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy3zzl, economic.getCy3zzl());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy3zjz, economic.getCy3zjz());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy3zzcy, economic.getCy3zzcy());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_cy3js, economic.getCy3js());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_dfczzsr, economic.getDfczzsr());
			this.addValue(put, famByt, HbaseTableCityEconomic.Column_Name_dfczzsrzzl, economic.getDfczzsrzzl());
			
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}

	/**
	 * 提交城市产业招商和布局信息
	 * 
	 * @param city
	 * @return status
	 */
	public HbaseStatus insertCitydustry(HBCityIndustry industry) {
		this.currentClientNum++;
		HbaseStatus status = HbaseStatus.successed;
		Table table = null;
		if(industry.getPostCode() == null || industry.getPostCode().length() == 0 || industry.getName() == null || industry.getName().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityIndustry.TABLE_NAME));
			Put put = new Put(Bytes.toBytes(industry.getPostCode() + "_ind_" + MD5Util.createMd5(industry.getName())));
			byte[] famByt = Bytes.toBytes(HbaseTableCityIndustry.COLUMN_FAMILY);
			
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_name, industry.getName());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_attractInvest, industry.getAttractInvest());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_plan, industry.getPlan());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_police, industry.getPolice());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_leader, industry.getLeader());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_cybj, industry.getCybj());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_advantage, industry.getAdvantage());
			this.addValue(put, famByt, HbaseTableCityIndustry.Column_Name_cyzc, industry.getCyzc());
			
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}

	/**
	 * 提交对比城市信息
	 * 
	 * @param city
	 * @return status
	 */
	public HbaseStatus insertCityCom(HBCityComparison compar) {
		this.currentClientNum++;
		HbaseStatus status = HbaseStatus.successed;
		Table table = null;
		if(compar.getPostCode() == null || compar.getPostCode().length() == 0 || compar.getComPostCode() == null || compar.getComPostCode().length() == 0){
			return HbaseStatus.illogic;
		}
		try {
			this.testConn();
			table = conn.getTable(this.tableNameMap.get(HbaseTableCityComparison.TABLE_NAME));
			Put put = new Put(Bytes.toBytes(compar.getPostCode() + "_com_" + compar.getComPostCode()));
			byte[] famByt = Bytes.toBytes(HbaseTableCityComparison.COLUMN_FAMILY);
			
			this.addValue(put, famByt, HbaseTableCityComparison.Column_Name_name, compar.getName());
			this.addValue(put, famByt, HbaseTableCityComparison.Column_Name_report, compar.getReport());
			
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
			status = HbaseStatus.exception;
		} finally{
			this.closeConn(table, conn);
		}
		this.currentClientNum--;
		return status;
	}
	
	/**
	 * 赋值
	 * @param put
	 * @param famByt
	 * @param qualifier
	 * @param value
	 */
	private void addValue(Put put, byte[] famByt, String qualifier, String value){
		if(value != null){
			put.addColumn(famByt, Bytes.toBytes(qualifier), Bytes.toBytes(value));
		}
	}

	/**
	 * 获取指定列的值
	 * 
	 * @param result
	 * @param familly
	 * @param rowName
	 * @return
	 */
	protected String getRow(Result result, String familly, String rowName) {
		byte[] byts = result.getValue(Bytes.toBytes(familly),
				Bytes.toBytes(rowName));
		String row = null;
		if (byts != null && byts.length > 0) {
			row = Bytes.toString(byts);
		}
		return row;
	}

	private void testConn(){
		if(this.conn == null){
			this.initConn();
		}
	}
	
	/*protected void close(ResultScanner rs, Table table) {
		try {
			rs.close();
		} catch (Exception e) {}
		try {
			table.close();
		} catch (Exception e1) {}
	}
	*/
	private void closeConn(){
		try {
			this.conn.close();
		} catch (Exception e) {}
		this.conn = null;
	}
	private void closeConn(Table table,Connection conn){
		try {
			table.close();
			conn.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	public void closeClient(){
		this.closeConn();
		try {
			this.timer.cancel();
		} catch (Exception e) {}
	}

}
